package com.xixi.canal.client.support;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Message对象解析工具类
 *
 * @author rewerma 2018-8-19 下午06:14:23
 * @version 1.0.0
 */
public class MessageUtil {

    public static List<Dml> parse4Dml(String destination, Message message) {

        if (null == message) return null;

        List<Dml> dmlList = new ArrayList<>(message.getEntries().size());

        for (CanalEntry.Entry entry : message.getEntries()) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            if (CanalEntry.EntryType.ROWDATA != entry.getEntryType()) continue;

            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;

            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                throw new RuntimeException("ERROR ## parser of event has an error, data:" + entry.toString(), e);
            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            final Dml dml = initDml(destination, message.getId(), entry, delayTime, rowChange, eventType);
            dmlList.add(dml);

            if (!rowChange.getIsDdl()) {
                addRowData(dml, eventType, rowChange.getRowDatasList());
            }
        }

        return dmlList;
    }

    private static void addRowData(Dml dml, CanalEntry.EventType eventType, List<CanalEntry.RowData> dataList) {
        for (CanalEntry.RowData rowData : dataList) {

            if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE
                    && eventType != CanalEntry.EventType.DELETE) {
                continue;
            }

            List<CanalEntry.Column> columnsList = null;
            Map<String, CanalEntry.Column> updateColumnMap = null;
            if (eventType == CanalEntry.EventType.DELETE) {
                columnsList = rowData.getBeforeColumnsList();
            } else if (eventType == CanalEntry.EventType.INSERT) {
                columnsList = rowData.getAfterColumnsList();
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                columnsList = rowData.getAfterColumnsList();
                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();

                updateColumnMap = beforeColumnsList
                        .stream()
                        .collect(
                        Collectors.toMap(x -> x.getName(), x -> x, (k1, k2) -> k1));

            }
            Dml.RowData row = new Dml.RowData();
            for (CanalEntry.Column column : columnsList) {
                Dml.ColumnData columnData = new Dml.ColumnData();
                columnData.setName(column.getName());
                columnData.setIsNull(column.getIsNull());
                columnData.setSqlType(column.getMysqlType());
                columnData.setUpdated(column.getUpdated());
                columnData.setPrimaryKey(column.getIsKey());

                if (column.getIsKey()) {
                    row.addKeyColumn(column.getName());
                }
                columnData.setValue(JdbcTypeUtil.typeConvert(dml.getTable(),
                        column.getName(),
                        column.getValue(),
                        column.getSqlType(),
                        column.getMysqlType()));
                if (null != updateColumnMap) {
                    CanalEntry.Column update = updateColumnMap.get(column.getName());
                    if (null != update) {
                        columnData.setOldValue(JdbcTypeUtil.typeConvert(dml.getTable(),
                                update.getName(),
                                update.getValue(),
                                update.getSqlType(),
                                update.getMysqlType()));
                    }
                }
                row.addColumn(columnData);
            }

            dml.addRowData(row);
        }
    }

    private static Dml initDml(String destination,
                               long batchId,
                               CanalEntry.Entry entry,
                               long delayTime,
                               CanalEntry.RowChange rowChange,
                               CanalEntry.EventType eventType) {
        final Dml dml = new Dml();
        dml.setDestination(destination);
        dml.setBatchId(batchId);

        dml.setDatabase(entry.getHeader().getSchemaName());
        dml.setTable(entry.getHeader().getTableName());
        dml.setEventType(eventType.toString());

        dml.setExecuteTime(entry.getHeader().getExecuteTime());
        dml.setDelayTime(delayTime);
        dml.setSyncTime(System.currentTimeMillis());

        dml.setIsDdl(rowChange.getIsDdl());
        dml.setSql(rowChange.getSql());
        return dml;
    }
}
